package com.myrpc.consumer.discovery.zk;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.myrpc.common.sessionchannel.Session;
import com.myrpc.consumer.netty.NettyClient;
import com.myrpc.consumer.xml.RpcInterFaceData;

import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class ChannelSessionManager {

	@Autowired
	NettyClient nettyClient;

	// 线程安全的Hashmap
	 ConcurrentHashMap<String, Map<String, Session>> channelSessions = new ConcurrentHashMap<String, Map<String, Session>>();

	
	
	public void watchNodeChannels(ConcurrentHashMap<String, String> nodesDatas) {
		
		//add
		if(channelSessions.size()<nodesDatas.size()) {
			
			Iterator<Entry<String, String>> iterator = nodesDatas.entrySet().iterator();
			
			while(iterator.hasNext()) {
				Entry<String, String> entry = iterator.next();
				
				if(isOnline(entry.getKey())) {
					continue;
				}else {
					log.info("key:{},value:{}",entry.getKey(), entry.getValue());
					this.createChannel(entry.getKey(), entry.getValue());
				}
				
			}
		}
		
		//delete
		else {
			Iterator<Entry<String, Map<String, Session>>> iterator = channelSessions.entrySet().iterator();
			while(iterator.hasNext()) {
				Entry<String, Map<String, Session>> entry = iterator.next();
				
				if(nodesDatas.containsKey(entry.getKey())) {
					continue;
				}else {
					removeChannel(entry.getKey());
				}
				
			}
			
		}
		
	}
	
	private void removeChannel(String key) {
		
		Map<String, Session> map = channelSessions.get(key);
		Iterator<Entry<String, Session>> iterator = map.entrySet().iterator();
		
		while(iterator.hasNext()) {
			Entry<String, Session> entry = iterator.next();
			Session session = entry.getValue();
			
			if(session.getChannel().isOpen()) {
				session.getChannel().disconnect();
			}
			
		}
		channelSessions.remove(key);
		log.info("remove:{}, size：{}",key,channelSessions.size());
	}

	public boolean isOnline(String key) {
		return channelSessions.containsKey(key);
	}
	
	public int size() {
		return channelSessions.size();
	}

	public void createChannel(String nodeName, String host) {

		String[] _str = host.split(":");

		Map<String, String> interfaceDatas = RpcInterFaceData.interfaceDatas;
		Iterator<Entry<String, String>> iterator = interfaceDatas.entrySet().iterator();
		Map<String, Session> sessionMap = new HashMap<String, Session>();
		while (iterator.hasNext()) {
			Entry<String, String> entry = iterator.next();
			Channel channel = nettyClient.connect(_str[0], Integer.parseInt(_str[1]));
			sessionMap.put(entry.getValue(), new Session(channel, entry.getValue(), entry.getKey()));
		}

		channelSessions.put(nodeName, sessionMap);
		log.info("channelSessions size{}", channelSessions.size());
	}

	public ConcurrentHashMap<String, Map<String, Session>> getChannelSession() {
		// TODO Auto-generated method stub
		return channelSessions;
	}


}
